package com.ibm.cps.demo;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Random;

import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

import com.ibm.cps.message.MessageFields;
import com.ibm.cps.newmessage.AbstractDataSourceMetadata;
import com.ibm.cps.newmessage.DataSourceMetadataFactory;
import com.ibm.cps.newmessage.MQDataSourceMetadata;
import com.ibm.interfaces.IMessageClient;
import com.ibm.interfaces.IPersistency;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.mongo.MongoPersistency;
import com.ibm.util.JSONUtility;
import com.ibm.util.LocalConfig;

/**
 * monitor source data message
 * 
 * @author Jingjing Wang 2015-3-23
 */
public class PublishSourceData {

	private static IPersistency persistency;
	private String tenantid;
	private Collection<String> dataSourceMetadata = null;
	private Logger logger;

	public PublishSourceData(String tenantid) throws Exception {
		this.tenantid = tenantid;
		persistency = MongoPersistency.getInstace();
		logger = Logger.getLogger(PublishSourceData.class);
		PropertyConfigurator.configure(LocalConfig.getLoggerProperties());
		dataSourceMetadata = persistency.getDataSource(tenantid, null);
	}

	public void publishMessage() throws Exception {
		logger.info("topics:"+ dataSourceMetadata);
		if (dataSourceMetadata == null || dataSourceMetadata.size() <= 0) {
			logger.warn("No data topics for user " + tenantid);
			return;
		}
		List<PublishDescriptor> descriptors = new ArrayList<>();
		for (String metaData : dataSourceMetadata) {
			String type = JSONUtility.getString(MessageFields.TYPE, metaData);
			AbstractDataSourceMetadata ds = DataSourceMetadataFactory.parseDataSourceMetadata(metaData, type);
			if (!(ds instanceof MQDataSourceMetadata)) {
				continue;
			}
			String topicName = ds.getSubscribeTarget();
			IMessageClient sender = ds.getDataSourceConsumerClient();
			ITimeSeriesSchema schema = ds.getTimeSeriesSchema();
			PseudoDataProducer producer = new PseudoDataProducer(schema);
			descriptors.add(new PublishDescriptor(ds.getName(), sender, topicName, producer));
		}

		int startValue = 5;
		int currentValue = startValue;
		if (dataSourceMetadata != null) {
			while (true) {
				for (PublishDescriptor descriptor : descriptors) {
					
					Thread.sleep(1000);	
					for(int i = 0;i<5;i++)
					{
						long indicator = new Random().nextLong() % 500;
						byte[] bytesToSend = descriptor.dataProducer.produce(i, indicator);
						descriptor.sender.publishMessage(descriptor.topicname, bytesToSend);
						logger.info("Send message: " + new String(bytesToSend) + " to datasource: " + descriptor.datasourceName);
					}
				}
			}
		}
	}

	private static final class PublishDescriptor {
		final String datasourceName;
		final IMessageClient sender;
		final String topicname;
		final PseudoDataProducer dataProducer;
		
		public PublishDescriptor(String datasourceName, IMessageClient sender,
				String topicname, PseudoDataProducer dataProducer) {
			this.datasourceName = datasourceName;
			this.sender = sender;
			this.topicname = topicname;
			this.dataProducer = dataProducer;
		}
		
	}

	public static void main(String[] args) throws Exception {
		if (args == null || args.length < 1) {
			System.err.println("Usage: Program <tenantid>");
			return;
		}
		String tenantid = args[0];
		PublishSourceData publishSourceData = new PublishSourceData(tenantid);
		publishSourceData.publishMessage();
	}
}
